package com.ydl.learning.flink.demo.pipeline.impl

import com.ydl.learning.flink.demo.pipeline.{BaseProcedureImpl, Process}
import org.apache.flink.api.common.state.MapStateDescriptor
import org.apache.flink.api.common.typeinfo.BasicTypeInfo
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.util.Collector
import org.codehaus.janino.ExpressionEvaluator
import org.slf4j.{Logger, LoggerFactory}

/**
 *
 *
 * @author ydl
 * @since 2022/1/7
 */
class BroadcastProcessImpl extends Process[String] with BaseProcedureImpl with Serializable {
  private final val log: Logger = LoggerFactory.getLogger(classOf[BroadcastProcessImpl])

  override def run: DataStream[String] = {
    val log_ = log
    log_.info("run")
    val streamEnv: StreamExecutionEnvironment = this.dataStream.executionEnvironment
    val broadcastStream = streamEnv.socketTextStream("localhost", 8888, '\n')
    val broadcastStateDescriptor: MapStateDescriptor[String, String] = new MapStateDescriptor[String, String](
      "broadcast",
      BasicTypeInfo.STRING_TYPE_INFO,
      BasicTypeInfo.STRING_TYPE_INFO
    )
    val broadcastConfig = broadcastStream.broadcast(broadcastStateDescriptor)
    this.dataStream.connect(broadcastConfig).process(process).name("process").map(_.toString)
  }

  def process: BroadcastProcessFunction[String, String, Object] = {
    val log_ = log
    new BroadcastProcessFunction[String, String, Object] {
      var ee: ExpressionEvaluator = _

      override def processElement(value: String,
                                  ctx: BroadcastProcessFunction[String, String, Object]#ReadOnlyContext,
                                  out: Collector[Object]): Unit = {
        log_.info("get data:{}", value)
        out.collect(ee.evaluate(value))
      }

      override def processBroadcastElement(value: String,
                                           ctx: BroadcastProcessFunction[String, String, Object]#Context,
                                           out: Collector[Object]): Unit = {
        log_.info("get rule:{}", value)
        ee = new ExpressionEvaluator()
        ee.setParameters(Array[String]("a", "b"), Array[Class[_]](classOf[String], classOf[String]))
        ee.setExpressionType(classOf[String])
        ee.setDefaultImports("java.lang.String")
        ee.cook(value)
      }
    }
  }
}

object BroadcastProcessImpl {
  def main(args: Array[String]): Unit = {
  }
}
